package com.baixiaowen.com.disruptor;

import com.baixiaowen.com.entity.TranslatorDataWapper;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;

public class RingBufferWorkerPoolFactory {

    private static class SingletonHolder {
        static final RingBufferWorkerPoolFactory instance = new RingBufferWorkerPoolFactory();
    }

    private RingBufferWorkerPoolFactory () {}

    public static RingBufferWorkerPoolFactory getInstance() {
        return SingletonHolder.instance;
    }

    private Map<String, MessageProducer> producers = new ConcurrentHashMap<>();

    private Map<String, MessageConsumer> consumers = new ConcurrentHashMap<>();

    private RingBuffer<TranslatorDataWapper> ringBuffer;

    private SequenceBarrier sequenceBarrier;

    private WorkerPool<?> workerPool;

    public void initAndStart(ProducerType producerType, int ringbufferSize, WaitStrategy waitStrategy, MessageConsumer[] messageConsumers) {
        // 1. 构建ringBuffer对象
        this.ringBuffer = RingBuffer.create(producerType,
                new EventFactory<TranslatorDataWapper>() {
                    @Override
                    public TranslatorDataWapper newInstance() {
                        return new TranslatorDataWapper();
                    }
                },
                ringbufferSize,
                waitStrategy);

        // 2.设置序号栅栏
        this.sequenceBarrier = this.ringBuffer.newBarrier();

        // 3. 设置工作池
        this.workerPool = new WorkerPool<TranslatorDataWapper>(this.ringBuffer,
                this.sequenceBarrier,
                new EventExceptionHandler(),
                messageConsumers);

        // 4. 把所构建的消费者置入池中
        for (MessageConsumer mc : messageConsumers) {
            this.consumers.put(mc.getConsumerId(), mc);
        }

        // 5. 添加Sequences
        this.ringBuffer.addGatingSequences(this.workerPool.getWorkerSequences());

        // 6. 启动工作池
        this.workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() / 2));
    }

    public MessageProducer getMessageProducer(String producerId) {
        MessageProducer messageProducer = this.producers.get(producerId);
        if (messageProducer == null) {
            messageProducer = new MessageProducer(producerId, this.ringBuffer);
            this.producers.put(producerId, messageProducer);
        }
        return messageProducer;
    }

    /**
     * 异常静态类
     */
    static class EventExceptionHandler implements ExceptionHandler {
        /**
         * 进行消费的时候的异常处理
         * @param ex
         * @param sequence
         * @param event
         */
        @Override
        public void handleEventException(Throwable ex, long sequence, Object event) {

        }

        /**
         * 启动的时候的异常处理
         * @param ex
         */
        @Override
        public void handleOnStartException(Throwable ex) {

        }

        /**
         * 结束的时候的异常处理
         * @param ex
         */
        @Override
        public void handleOnShutdownException(Throwable ex) {

        }
    }

}
